package redis.clients.jedis;

import java.io.Closeable;
import java.io.IOException;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;

import javax.net.ssl.HostnameVerifier;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSocketFactory;

import redis.clients.jedis.commands.ProtocolCommand;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisDataException;
import redis.clients.jedis.util.IOUtils;
import redis.clients.jedis.util.RedisInputStream;
import redis.clients.jedis.util.RedisOutputStream;
import redis.clients.jedis.util.SafeEncoder;


/**
 * 连接
 */
public class Connection implements Closeable {

    private static final byte[][] EMPTY_ARGS = new byte[0][];


    /**
     * Jedis Connection中的socketParamModified是一个布尔值，用于表示是否修改了连接的socket参数。
     * 当使用Jedis连接Redis时，可以通过设置socket参数来优化连接性能。
     * 例如，可以设置TCP_NODELAY选项来禁用Nagle算法，从而减少数据传输的延迟。
     * 如果socketParamModified为true，则表示在连接过程中修改了socket参数；
     * 如果为false，则表示没有修改socket参数。这个值通常用于调试和监控目的，以了解连接的性能和状态。
     */
    private boolean socketParamModified = false; // for backward compatibility

    /**
     * JedisSocketFactory 是一个用于创建 Jedis 连接的工厂类。
     * 它负责将网络连接封装成 Jedis 对象，以便在应用程序中使用。
     * JedisSocketFactory 通常与 JedisPool 一起使用，以实现连接池的功能。
     */
    private JedisSocketFactory socketFactory; // TODO: should be final


    private Socket socket;
    private RedisOutputStream outputStream;
    private RedisInputStream inputStream;


    private int soTimeout = Protocol.DEFAULT_TIMEOUT;
    private int infiniteSoTimeout = 0;
    private boolean broken = false;

    public Connection() {
        this(Protocol.DEFAULT_HOST, Protocol.DEFAULT_PORT);
    }

    /**
     * @param host
     * @deprecated This constructor will be removed in future. It can be replaced with
     * {@link #Connection(java.lang.String, int)} with the host and {@link Protocol#DEFAULT_PORT}.
     */
    @Deprecated
    public Connection(final String host) {
        this(host, Protocol.DEFAULT_PORT);
    }

    public Connection(final String host, final int port) {
        this(new HostAndPort(host, port), DefaultJedisClientConfig.builder().build());
    }

    /**
     * @deprecated This constructor will be removed in future.
     */
    @Deprecated
    public Connection(final String host, final int port, final boolean ssl) {
        this(new HostAndPort(host, port), DefaultJedisClientConfig.builder().ssl(ssl).build());
    }

    /**
     * @deprecated This constructor will be removed in future.
     */
    @Deprecated
    public Connection(final String host, final int port, final boolean ssl,
                      SSLSocketFactory sslSocketFactory, SSLParameters sslParameters,
                      HostnameVerifier hostnameVerifier) {
        this(new HostAndPort(host, port), DefaultJedisClientConfig.builder().ssl(ssl)
                .sslSocketFactory(sslSocketFactory).sslParameters(sslParameters)
                .hostnameVerifier(hostnameVerifier).build());
    }

    public Connection(final HostAndPort hostAndPort, final JedisClientConfig clientConfig) {
        this(new DefaultJedisSocketFactory(hostAndPort, clientConfig));
        this.soTimeout = clientConfig.getSocketTimeoutMillis();
        this.infiniteSoTimeout = clientConfig.getBlockingSocketTimeoutMillis();
    }


    /**
     * JedisFactory --> makeObject 会创建Jedis实例
     */
    public Connection(final JedisSocketFactory jedisSocketFactory) {
        this.socketFactory = jedisSocketFactory;
        this.soTimeout = jedisSocketFactory.getSoTimeout();
    }

    @Override
    public String toString() {
        return "Connection{" + socketFactory + "}";
    }

    public Socket getSocket() {
        return socket;
    }

    public int getConnectionTimeout() {
        return socketFactory.getConnectionTimeout();
    }

    public int getSoTimeout() {
        return soTimeout;
    }

    /**
     * @param connectionTimeout
     * @deprecated This method is not supported anymore and is kept for backward compatibility. It
     * will be removed in future.
     */
    @Deprecated
    public void setConnectionTimeout(int connectionTimeout) {
        socketFactory.setConnectionTimeout(connectionTimeout);
    }

    public void setSoTimeout(int soTimeout) {
        socketFactory.setSoTimeout(soTimeout);
        this.soTimeout = soTimeout;
        if (this.socket != null) {
            try {
                this.socket.setSoTimeout(soTimeout);
            } catch (SocketException ex) {
                broken = true;
                throw new JedisConnectionException(ex);
            }
        }
    }

    public void setInfiniteSoTimeout(int infiniteSoTimeout) {
        this.infiniteSoTimeout = infiniteSoTimeout;
    }

    public void setTimeoutInfinite() {
        try {
            if (!isConnected()) {
                connect();
            }
            socket.setSoTimeout(infiniteSoTimeout);
        } catch (SocketException ex) {
            broken = true;
            throw new JedisConnectionException(ex);
        }
    }

    public void rollbackTimeout() {
        try {
            socket.setSoTimeout(socketFactory.getSoTimeout());
        } catch (SocketException ex) {
            broken = true;
            throw new JedisConnectionException(ex);
        }
    }

    public void sendCommand(final ProtocolCommand cmd, final String... args) {
        final byte[][] bargs = new byte[args.length][];
        for (int i = 0; i < args.length; i++) {
            bargs[i] = SafeEncoder.encode(args[i]);
        }
        sendCommand(cmd, bargs);
    }

    public void sendCommand(final ProtocolCommand cmd) {
        sendCommand(cmd, EMPTY_ARGS);
    }


    /**
     * 用于向redis服务器发送原始命令。该方法接收byte数组作为参数，该数组包含了要发送的命令及参数
     */
    public void sendCommand(final ProtocolCommand cmd, final byte[]... args) {
        try {
            connect();
            Protocol.sendCommand(outputStream, cmd, args);
        } catch (JedisConnectionException ex) {
            /*
             * When client send request which formed by invalid protocol, Redis send back error message
             * before close connection. We try to read it to provide reason of failure.
             */
            try {
                String errorMessage = Protocol.readErrorLineIfPossible(inputStream);
                if (errorMessage != null && errorMessage.length() > 0) {
                    ex = new JedisConnectionException(errorMessage, ex.getCause());
                }
            } catch (RuntimeException e) {
                /*
                 * Catch any JedisConnectionException occurred from InputStream#read and just
                 * ignore. This approach is safe because reading error message is optional and connection
                 * will eventually be closed.
                 */
            }
            // Any other exceptions related to connection?
            broken = true;
            throw ex;
        }
    }

    public String getHost() {
        return socketFactory.getHost();
    }

    /**
     * @param host
     * @deprecated This method will be removed in future.
     */
    @Deprecated
    public void setHost(final String host) {
        socketFactory.setHost(host);
        socketParamModified = true;
    }

    public int getPort() {
        return socketFactory.getPort();
    }

    /**
     * @param port
     * @deprecated This method will be removed in future.
     */
    @Deprecated
    public void setPort(final int port) {
        socketFactory.setPort(port);
        socketParamModified = true;
    }


    /**
     * jedis连接redis
     */
    public void connect() throws JedisConnectionException {
        if (socketParamModified) { // this is only for backward compatibility
            try {
                disconnect();
            } catch (RuntimeException e) {
                // swallow
            }
        }
        if (!isConnected()) {
            try {
                // 创建socket连接
                socket = socketFactory.createSocket();

                outputStream = new RedisOutputStream(socket.getOutputStream());
                inputStream = new RedisInputStream(socket.getInputStream());
            } catch (IOException ioe) {
                broken = true;
                throw new JedisConnectionException("Failed to create input/output stream", ioe);
            } catch (JedisConnectionException jce) {
                broken = true;
                throw jce;
            } finally {
                // 连接创建失败,最终关闭socker
                if (broken) {
                    IOUtils.closeQuietly(socket);
                }
            }
        }
    }

    @Override
    public void close() {
        disconnect();
    }

    /**
     * Close the socket and disconnect the server.
     */
    public void disconnect() {
        if (isConnected()) {
            try {
                outputStream.flush();
                socket.close();
            } catch (IOException ex) {
                broken = true;
                throw new JedisConnectionException(ex);
            } finally {
                IOUtils.closeQuietly(socket);
            }
        }
    }

    public boolean isConnected() {
        return socket != null
                // 检查套接字是否已绑定到本地地址和端口的方法。
                // 如果套接字已绑定，则返回 True，否则返回 False
                && socket.isBound()
                // 检查套接字是否已关闭的方法。
                // 如果套接字已关闭，则返回 True，否则返回 False
                && !socket.isClosed()
                // 检查套接字是否已连接的方法。
                // 如果套接字已连接，则返回 True，否则返回 False
                && socket.isConnected()
                // 用于检查套接字输入流是否已关闭的方法。
                // 如果输入流已关闭，则返回 True，否则返回 False
                && !socket.isInputShutdown()
                // 用于检查套接字输出流是否已关闭的方法。
                // 如果输出流已关闭，则返回 True，否则返回 False
                && !socket.isOutputShutdown();
    }

    public String getStatusCodeReply() {
        flush();
        final byte[] resp = (byte[]) readProtocolWithCheckingBroken();
        if (null == resp) {
            return null;
        } else {
            return SafeEncoder.encode(resp);
        }
    }

    public String getBulkReply() {
        final byte[] result = getBinaryBulkReply();
        if (null != result) {
            return SafeEncoder.encode(result);
        } else {
            return null;
        }
    }

    public byte[] getBinaryBulkReply() {
        flush();
        return (byte[]) readProtocolWithCheckingBroken();
    }

    public Long getIntegerReply() {
        flush();
        return (Long) readProtocolWithCheckingBroken();
    }

    public List<String> getMultiBulkReply() {
        return BuilderFactory.STRING_LIST.build(getBinaryMultiBulkReply());
    }

    @SuppressWarnings("unchecked")
    public List<byte[]> getBinaryMultiBulkReply() {
        flush();
        return (List<byte[]>) readProtocolWithCheckingBroken();
    }

    @Deprecated
    public List<Object> getRawObjectMultiBulkReply() {
        return getUnflushedObjectMultiBulkReply();
    }

    @SuppressWarnings("unchecked")
    public List<Object> getUnflushedObjectMultiBulkReply() {
        return (List<Object>) readProtocolWithCheckingBroken();
    }

    public List<Object> getObjectMultiBulkReply() {
        flush();
        return getUnflushedObjectMultiBulkReply();
    }

    @SuppressWarnings("unchecked")
    public List<Long> getIntegerMultiBulkReply() {
        flush();
        return (List<Long>) readProtocolWithCheckingBroken();
    }

    public Object getOne() {
        flush();
        return readProtocolWithCheckingBroken();
    }

    public boolean isBroken() {
        return broken;
    }

    protected void flush() {
        try {
            outputStream.flush();
        } catch (IOException ex) {
            broken = true;
            throw new JedisConnectionException(ex);
        }
    }

    protected Object readProtocolWithCheckingBroken() {
        if (broken) {
            throw new JedisConnectionException("Attempting to read from a broken connection");
        }

        try {
            return Protocol.read(inputStream);
        } catch (JedisConnectionException exc) {
            broken = true;
            throw exc;
        }
    }

    public List<Object> getMany(final int count) {
        flush();
        final List<Object> responses = new ArrayList<>(count);
        for (int i = 0; i < count; i++) {
            try {
                responses.add(readProtocolWithCheckingBroken());
            } catch (JedisDataException e) {
                responses.add(e);
            }
        }
        return responses;
    }
}
